package com.lagou.rpc.consumer;

import com.lagou.rpc.consumer.client.RpcClient;
import com.lagou.rpc.consumer.client.lb.ClientLoadBalancer;
import org.apache.curator.framework.CuratorFramework;
import org.apache.curator.framework.recipes.cache.CuratorCache;
import org.apache.curator.framework.recipes.cache.CuratorCacheListener;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheEvent;
import org.apache.curator.framework.recipes.cache.PathChildrenCacheListener;
import org.apache.curator.utils.ZKPaths;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.boot.CommandLineRunner;
import org.springframework.boot.SpringApplication;
import org.springframework.boot.autoconfigure.SpringBootApplication;

/**
 * @author xingyi
 */
@SpringBootApplication
public class ClientBootstrapApplication implements CommandLineRunner {

    private static final Logger logger = LoggerFactory.getLogger(ClientBootstrapApplication.class);

    @Autowired
    private ClientLoadBalancer clientLoadBalancer;

    @Autowired
    private CuratorFramework curatorFramework;

    public static void main(String[] args) {
        SpringApplication.run(ClientBootstrapApplication.class, args);
    }

    @Override
    public void run(String... args) throws Exception {
        runWithClientBoadBalancer();
    }

    private void runWithClientBoadBalancer() throws Exception {

        final CuratorCache curatorCache = CuratorCache.builder(curatorFramework, "/rpcserver").build();
        final CuratorCacheListener cacheListener = CuratorCacheListener.builder().forPathChildrenCache("/rpcserver", curatorFramework, new PathChildrenCacheListener() {
            @Override
            public void childEvent(CuratorFramework client, PathChildrenCacheEvent event) throws Exception {
                if (event.getData() == null) {
                    logger.info("event type:{}", event.getType());
                    return;
                }
                final String server = ZKPaths.getNodeFromPath(event.getData().getPath());
                switch (event.getType()) {
                    case CHILD_ADDED:
                        logger.info("server added:{}", server);
                        if (!clientLoadBalancer.contains(server)) {
                            addRpcClient(server);
                            clientLoadBalancer.onUpdateStat(server, event.getData().getData());
                        }
                        break;
                    case CHILD_REMOVED:
                        logger.info("server removed:{}", server);
                        clientLoadBalancer.removeClient(server);
                        break;
                    case CHILD_UPDATED:
                        logger.info("server updated:{}", server);
                        clientLoadBalancer.onUpdateStat(server, event.getData().getData());
                        break;
                    default:
                        logger.info("event type:{}", event.getType());
                }
            }
        }).build();
        curatorCache.listenable().addListener(cacheListener);
        curatorCache.start();
    }

    private void addRpcClient(String server) {
        String[] parts = server.split(":");
        final RpcClient rpcClient = new RpcClient(parts[0], Integer.parseInt(parts[1]));
        rpcClient.setClientLoadBalancer(clientLoadBalancer);
        rpcClient.setCuratorFramework(curatorFramework);
        clientLoadBalancer.addClient(rpcClient);
        rpcClient.reportStat();
    }
}
